#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
File: vm_agent.py
Time: 2025/05/27 17:59
Description: FastAPI router for handling VM agent game start and end requests.
"""

import asyncio
import json
import os
from typing import Dict, Optional, List
from datetime import datetime
from enum import IntEnum
from pydantic import BaseModel, validator, ValidationError
from fastapi import Request, HTTPException, APIRouter

from apps.mock.models.vm_agent import StartGameReq, GameStatus, GsmGameStateNotify, CgVmAgentGameEventNotify, \
    game_event_to_string, GameEvent, CallGsmStartGameInfo, EndGameReq, CgAgentResp, Ret
from utils.kafka_util.kafka_client import KafkaProducer
from utils.logging_tool.log_control import INFO,ERROR,WARNING

vm_agent = APIRouter(prefix = "/{prefix:path}/v1/cgagent/game")


# GameStatus enum



# Task manager for background tasks
class TaskManager:
    """Manages background tasks with asyncio Events for stopping tasks."""

    def __init__(self):
        self.tasks: Dict[str, asyncio.Event] = {}
        self.lock = asyncio.Lock()

    async def add_task(self, task_id: str, event: asyncio.Event) -> None:
        """Add a task to the manager."""
        async with self.lock:
            self.tasks[task_id] = event
            INFO.logger(f"added task {task_id}")

    async def remove_task(self, task_id: str) -> None:
        """Remove a task from the manager."""
        async with self.lock:
            if task_id in self.tasks:
                self.tasks[task_id].set()
                del self.tasks[task_id]
                INFO.logger(f"Removed task {task_id}")


manager = TaskManager()
kafka_producer = KafkaProducer()

async def send_kafka_message(topic: str, message: str) -> None:
    """Send a message to a Kafka topic with error handling."""
    try:
        await kafka_producer.send_message(topic, message)
    except Exception as e:
        ERROR.logger(f"Failed to send Kafka message to {topic}: {str(e)}")


def create_game_state_notify(req: StartGameReq, state: GameStatus, timestamp: datetime = None) -> GsmGameStateNotify:
    """Create a GsmGameStateNotify object with common fields."""
    return GsmGameStateNotify(
        gid = int(req.gid) if req.gid else None,
        nonce = req.flowId,
        state = state,
        timestamp = (timestamp or datetime.now().astimezone()).isoformat(),
        uid = int(req.uuid) if req.uuid else None,
        device_id = req.devId,
        area_type = req.areaType,
        vmid = req.vmid
    )


async def process_start_game_async(req: StartGameReq):
    try:
        # Pre-create all messages to minimize async context switching
        timestamp = datetime.now().astimezone()
        uid_str = str(req.uid or 0)
        topic = f"game_operate_report_{uid_str[-1] if uid_str else '0'}"

        # Create all notifications in one batch
        notifications = [
            CgVmAgentGameEventNotify(
                nonce = req.flowId,
                device_id = req.devId,
                uid = int(req.uid) if req.uid else None,
                vmid = req.vmid,
                gid = int(req.gid) if req.gid else None,
                event = game_event_to_string[GameEvent.GameEventCallGsmStartGameSuccess],
                timestamp = timestamp,
                msg = "startGame success",
                version = 1,
                gsmStartGame = CallGsmStartGameInfo(),
                scheduleId = req.scheduleId
            ),
            create_game_state_notify(req, GameStatus.GameStatusStartProcing, timestamp),
            create_game_state_notify(req, GameStatus.GameStatusStartSuccess, timestamp),
            create_game_state_notify(req, GameStatus.GameStatusReadyRender, timestamp)
        ]

        # Single gather call for all Kafka sends
        await asyncio.gather(
            send_kafka_message("cgvmagent_game_event_report", notifications[0].json()),
            *[send_kafka_message(topic, n.json()) for n in notifications[1:]]
        )

        task_id = f"{req.uid}_{req.flowId}"
        stop_event = asyncio.Event()

        async def task_runner():
            try:
                async with asyncio.timeout(3600):
                    while not stop_event.is_set():
                        notify = create_game_state_notify(req, GameStatus.GameStatusUserInActive)
                        await send_kafka_message(topic, notify.json())
                        await asyncio.sleep(int(os.getenv("USER_INACTIVE_INTERVAL", 8)))
            except asyncio.TimeoutError:
                ERROR.logger(f"Task {task_id} timed out")
            except Exception as e:
                ERROR.logger(f"Task {task_id} failed: {str(e)}")
            finally:
                try:
                    notify = create_game_state_notify(req, GameStatus.GameStatusCloseSuccess)
                    await asyncio.gather(
                        send_kafka_message("cgvmagent_game_event_report", notify.json())
                    )
                except Exception as e:
                    ERROR.logger(f"Failed to send final notification: {str(e)}")
                finally:
                    await manager.remove_task(task_id)

        await manager.add_task(task_id, stop_event)
        asyncio.create_task(task_runner(), name = f"game_monitor_{task_id}")
    except Exception as e:
        ERROR.logger(f"Async processing failed: {str(e)}")


@vm_agent.post("/startGame", status_code = 200)
async def cg_agent_start_game(request: Request):
    full_path = request.url.path
    body = await request.body()
    INFO.logger(f"Received request: {full_path}, body: {body.decode()}")

    try:
        req = StartGameReq(**await request.json())
    except (json.JSONDecodeError, ValidationError) as e:
        ERROR.logger(f"Failed to parse request body: {str(e)}")
        raise HTTPException(
            status_code = 400,
            detail = {"ret": {"code": -1, "msg": "Invalid request body", "reason": str(e)}}
        )

    # Create and start background task
    task_id = f"{req.uid}_{req.flowId}"
    stop_event = asyncio.Event()
    asyncio.create_task(process_start_game_async(req), name = f"game_start_{task_id}")
    await manager.add_task(task_id, stop_event)

    resp = CgAgentResp(ret = Ret(code = 0, msg = "Success", reason = ""))
    return resp


@vm_agent.post("/endGame", status_code = 200)
async def cg_agent_end_game(request: Request):
    body = await request.body()
    INFO.logger(f"Received endGame request: {request.url.path}, body: {body.decode()}")

    try:
        req = EndGameReq(**await request.json())
    except (json.JSONDecodeError, ValidationError) as e:
        ERROR.logger(f"Failed to parse request body: {str(e)}")
        return CgAgentResp(ret = Ret(code = -1, msg = "Invalid request body", reason = str(e)))

    # Send game event notification
    event_notify = CgVmAgentGameEventNotify(
        nonce = req.flowId,
        uid = req.uid,
        vmid = req.vmid,
        gid = req.gid,
        event = game_event_to_string[GameEvent.GameEventCallGsmEndGameSuccess],
        timestamp = datetime.now().astimezone(),
        msg = "Game ended successfully",
        version = 1,
        scheduleId = req.scheduleId
    )

    await asyncio.gather(
        send_kafka_message("cgvmagent_game_event_report", event_notify.json())
    )

    # Stop the background task
    task_id = f"{req.uid}_{req.flowId}"
    await manager.remove_task(task_id)

    return CgAgentResp(ret = Ret(code = 0, msg = "Success", reason = ""))
